package com.bw.gmall.realtime.app.dwd;

import com.bw.gmall.realtime.utils.MyKafkaUtil;
import com.bw.gmall.realtime.utils.MysqlUtil;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

import java.time.Duration;

public class DwdTradeRefundPaySuc {
    public static void main(String[] args) {
        //todo 1.环境准备
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tenv = StreamTableEnvironment.create(env);
        //获取配置对象
        Configuration configuration = tenv.getConfig().getConfiguration();
        //过期时间是10秒
//        tenv.getConfig().setIdleStateRetention(Duration.ofSeconds(5));
        //为表关联时状态中存储的数据设置过期时间
//        configuration.setString("table.exec.state.ttl","5 s");

        //todo 2.状态后端设置

        //todo 3. 从kafka读取topic_db数据，封装为Flink SQL表
        tenv.executeSql("create table topic_db("+
                " `database` string,"+
                " `table` string,"+
                " `type` string,"+
                " `data` map<string,string>,"+
                " `old` map<string,string>,"+
                " `proc_time` as PROCTIME(),"+
                " `ts` string"+
                ")"+ MyKafkaUtil.getKafkaDDL("topic_db","refund_pay_suc_211126"));

        //todo 4.建立MySQL-LookUp 字典表
        tenv.executeSql(MysqlUtil.getBaseDicLookUpDDL());

        //todo 5. 读取退款表数据,并筛选退款成功数据
        Table refundPayment = tenv.sqlQuery("select " +
                " data['id'] id, " +
                " data['order_id'] order_id, " +
                " data['sku_id'] sku_id, " +
                " data['payment_type'] payment_type, " +
                " data['callback_time'] callback_time, " +
                " data['total_amount'] total_amount, " +
                " proc_time, " +
                " ts " +
                " from topic_db " +
                " where `table`='refund_payment' " +
                " and (`type`='update' or `type`='insert')" +
                " and data['refund_status']='0705' ");
        tenv.createTemporaryView("refund_payment",refundPayment);

        //todo 6.读取订单表数据并过滤退款成功订单数据
        Table orderInfo = tenv.sqlQuery("select" +
                " data['id'] id, " +
                " data['user_id'] user_id, " +
                " data['province_id'] province_id, " +
                " `old`" +
                " from topic_db " +
                " where `table`='order_info' " +
                " and (`type`='update' or `type`='insert') " +
                " and data['order_status']='1006' " +
                " and `old`['order_status'] is not null");
        tenv.createTemporaryView("order_info",orderInfo);


        //todo 7.读取退单表数据并过滤退款成功数据
        Table orderRefundInfo = tenv.sqlQuery("select " +
                " data['order_id'] order_id, " +
                " data['sku_id'] sku_id, " +
                " data['refund_num'] refund_num, " +
                " `old` " +
                " from topic_db " +
                " where `table`='order_refund_info' " +
                " and (`type`='update' or `type`='insert') " +
                " and data['refund_status']='0705' ");

        tenv.createTemporaryView("order_refund_info",orderRefundInfo);


        //todo 8.关联4张表获得退款成功表

        Table resultTable = tenv.sqlQuery("select " +
                " rp.id, " +
                " oi.user_id, " +
                " rp.order_id, " +
                " rp.sku_id, " +
                " oi.province_id, " +
                " rp.payment_type, " +
                " dic.dic_name payment_type_name, " +
                " date_format(rp.callback_time,'yyyy-MM-dd') date_id, " +
                " rp.callback_time, " +
                " ri.refund_num, " +
                " rp.total_amount, " +
                " rp.ts, " +
                " current_row_timestamp() row_op_ts " +
                " from refund_payment rp " +
                " join " +
                " order_info oi " +
                " on rp.order_id=oi.id " +
                " join " +
                " order_refund_info ri " +
                " on rp.order_id=ri.order_id " +
                " and rp.sku_id=ri.sku_id " +
                " join " +
                " base_dic for system_time as of rp.proc_time as dic " +
                " on rp.payment_type=dic.dic_code ");
        tenv.createTemporaryView("result_table",resultTable);



        //todo 9.创建kafka-Connector dwd_trade_refund_pay_suc表

        tenv.executeSql("create table dwd_trade_refund_pay_suc("+
                " id string, "+
                " user_id string, "+
                " order_id string, "+
                " sku_id string, "+
                " province_id string, "+
                " payment_type_code string, "+
                " payment_type_name string, "+
                " date_id string, "+
                " callback_time string, "+
                " refund_num string, "+
                " refund_amount string, "+
                " ts string, "+
                " row_op_ts timestamp_ltz(3)"+")"
                +MyKafkaUtil.getKafkaSinkDDL("dwd_trade_refund_pay_suc"));
        tenv.executeSql("select * from result_table").print();
        //todo 10.  将关联结果写入Kafka-Connector表
        tenv.executeSql("insert into dwd_trade_refund_pay_suc select * from result_table");




    }
}
